package com.wis;

import com.wis.bolt.common.*;
import com.wis.bolt.mpg.*;
import com.wis.bolt.risk.*;
import com.wis.bolt.to_kafka.*;
import com.wis.spout.LocalSpout;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.topology.TopologyBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.Properties;

/**
 * @Description: traffic_fan    主函数
 * @author: fan
 * @Date: Created in 2018/10/9 14:33
 * @Modified By:
 */
public class TopologyDrive {
    // 引入log4j
    private static final Logger LOGGER = LoggerFactory.getLogger(TopologyDrive.class);
    public static final String REDIS_KAZHI = "redis.key.kazhi";
    public static final String REDIS_POINT_KAZHI = "redis.key.point.kazhi";
    public static final String REDIS_MPG_KAZHI = "redis.key.mpg.kazhi";
    public static final String REDIS_RISK_REV = "redis.key.risk.revision";
    public static String configFileName;
    // 产品类型
    public static final String TYPE_PRODUCE = "type.producer";
    // top的名称
    private final static String TOPOLOGY_NAME = "topology.name";
    private final static String DEFAULT_TOPOLOGY_NAME = System.currentTimeMillis() + "_" + "topology_work";
    // top的work数
    private final static String TOP_WORKER_NUMBER = "top.worker.number";
    private final static String DEFAULT_TOP_WORKER_NUMBER = "3";
    // kafka生产者的连接
    public final static String KAFKA_PRODUCER_SERVER = "kafka.producer.server";
    // kafka消费的topic
    private final static String KAFKA_CONSUMER_TOPIC = "kafka.consumer.topic";
    // kafka消费者的group id
    private final static String KAFKA_CONSUMER_GROUP_ID = "kafka.consumer.group.id";
    //kafka消费者的连接
    private final static String KAFKA_CONSUMER_SERVER = "kafka.consumer.server";
    // kafka生产者的topic，卡值之后的数据传回
    public final static String KAFKA_PRODUCER_BEAN_TOPIC = "kafka.producer.bean.topic";
    // kafka生产者的topic，路段数据处理之后的数据传回---线
    public final static String KAFKA_PRODUCER_MPG_Line_TOPIC = "kafka.producer.mpg.line.topic";
    public final static String KAFKA_PRODUCER_MPG_Point_TOPIC = "kafka.producer.mpg.point.topic";
    // kafka生产者的topic，路段数据处理之后的数据传回---点
    public final static String KAFKA_PRODUCER_RISK_TOPIC = "kafka.producer.risk.topic";
    // kafka生产者的topic，综合指标点数据的传回
    public final static String KAFKA_PRODUCER_CD_POINT_TOPIC = "kafka.producer.cd.point.topic";
    // kafka生产者的topic，综合指标线数据的传回
    public static final String KAFKA_PRODUCER_CD_LINK_TOPIC = "kafka.producer.cd.link.topic";
    // kafka生产者的topic，风险告警信息线数据的传回
    public static final String KAFKA_PRODUCER_RISK_LINK_TOPIC = "kafka.producer.risk.link.topic";
    // kafka生产者的topic，风险告警信息点数据的传回
    public static final String KAFKA_PRODUCER_RISK_POINT_TOPIC = "kafka.producer.risk.point.topic";


    public static void main(String[] args) {
        if (args.length < 1)
            throw new RuntimeException("******please input filename**************");
        // 配置文件的名称
        configFileName = args[0];
        // String configFileName = "twn.properties";
        Properties properties = getProperties(configFileName);

        LOGGER.info(new Date(System.currentTimeMillis()) + "：开始执行2019-10-23的topology任务：");
        // Properties properties = TrafficUtils.getProp();
        TopologyBuilder builder = new TopologyBuilder();
        // storm从kafka拉取数据的配置列表
        // 两种不同的配置文件写法：省略掉了传统的kafka消费者的那种写法
        KafkaSpoutConfig.Builder kafkaBuilder = KafkaSpoutConfig.builder(properties.getProperty(KAFKA_CONSUMER_SERVER), properties.getProperty(KAFKA_CONSUMER_TOPIC))
                .setOffsetCommitPeriodMs(10000)//控制spout多久向kafka提交一次offset
                .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_LATEST)//最近未提交,若没有偏移量记录，从最新的数据开始消费
                // .setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)//最近未提交，若没有偏移量记录，从头开始消费
                .setProp(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_CONSUMER_GROUP_ID)
                .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
                .setProp(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000)
                .setProp(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000)
                // .setProp(ConsumerConfig.RECEIVE_BUFFER_CONFIG, 33554432)
                .setProp(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1000)
                .setProp(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class)
                .setProp(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 不同数据源采用不用的任务数量
        int type = Integer.parseInt(properties.getProperty("type.producer"));
        int riskLine = 5;
        int riskLineJiHe = 2;
        // int riskDian = 5;
        // int riskDianJiHe = 2;
        int mpgLine = 3;
        int mpgLineJiHe = 3;
        int mpgDian = 5;
        int mpgDianJiHe = 5;
        int toKafka = 1;

        if (type == 3) {
            riskLine = 20;
            riskLineJiHe = 10;
            // riskDian = 5;
            // riskDianJiHe = 2;
            mpgLine = 40;
            mpgLineJiHe = 40;
            mpgDian = 40;
            mpgDianJiHe = 40;
            toKafka = 10;
        }


        KafkaSpout kafkaSpout = new KafkaSpout(kafkaBuilder.build());
        builder.setSpout("kafkaSpoutToStorm", kafkaSpout, 1);// 数据源
        // builder.setSpout("kafkaSpoutToStorm", new LocalSpout(), 1);// 数据源
        // builder.setSpout("kafkaSpoutToStorm", new KafkaTwoSpout(), 1);// 自己写的kafka的接收端
        // 按桩点切
        builder.setBolt("ZhuangDianSpiltBolt", new ZhuangDianSpiltBolt(), 1).allGrouping("kafkaSpoutToStorm");
        // 按省份切割
        builder.setBolt("PACSplitBolt", new PACSplitBolt(), 1).allGrouping("ZhuangDianSpiltBolt");
        // 线
        builder.setBolt("QiePianXianBolt", new QiePianXianBolt(), 1).allGrouping("PACSplitBolt");
        builder.setBolt("YaoSuZuHeXianBolt", new YaoSuZuHeXianBolt(), 1).shuffleGrouping("QiePianXianBolt");
        // 点
        builder.setBolt("QiePianDianBolt", new QiePianDianBolt(), 1).allGrouping("PACSplitBolt");
        builder.setBolt("YaoSuZuHeDianBolt", new YaoSuZuHeDianBolt(), 1).shuffleGrouping("QiePianDianBolt");
        // 风险线
        builder.setBolt("RiskKaZhiXianBolt", new RiskKaZhiBolt(), 1).allGrouping("YaoSuZuHeXianBolt");
        builder.setBolt("RiskShouJiXianBolt", new RiskShouJiBolt(), 1).allGrouping("RiskKaZhiXianBolt");
        builder.setBolt("RiskJiHeToXianDanJiHeBolt", new RISKJiHeDanJiHeBolt(), riskLine).shuffleGrouping("RiskShouJiXianBolt");
        builder.setBolt("RiskMagicToLineBolt", new MagicToLineBolt(), riskLineJiHe).shuffleGrouping("RiskJiHeToXianDanJiHeBolt");
        // 风险点
        builder.setBolt("RiskKaZhiDianBolt", new RiskKaZhiBolt(), 1).allGrouping("YaoSuZuHeDianBolt");
        builder.setBolt("RiskShouJiDianBolt", new RiskShouJiBolt(), 1).allGrouping("RiskKaZhiDianBolt");
        builder.setBolt("RiskJiHeToDianDanJiHeBolt", new RISKJiHeDanJiHeBolt(), riskLine).shuffleGrouping("RiskShouJiDianBolt");
        builder.setBolt("RiskMagicToDianBolt", new MagicToDianBolt(), riskLineJiHe).shuffleGrouping("RiskJiHeToDianDanJiHeBolt");
        // 风险告警
        builder.setBolt("RiskLineDataBolt", new RiskLineDataBolt(), 1).allGrouping("RiskMagicToLineBolt");
        // 综合指标线
        builder.setBolt("ZongHeZuHeXianBolt", new ZongHeZuHeBolt(), 1).allGrouping("RiskKaZhiXianBolt");
        builder.setBolt("ZongHeZhuanXianBolt", new ZongHeZhuanBolt(), 1).shuffleGrouping("ZongHeZuHeXianBolt");
        builder.setBolt("ZongHeMagicToLineBolt", new ZongHeMagicToLineBolt(), 1).allGrouping("ZongHeZhuanXianBolt");
        // 综合指标点
        builder.setBolt("ZongHeZuHeDianBolt", new ZongHeZuHeBolt(), 1).allGrouping("RiskKaZhiDianBolt");
        builder.setBolt("ZongHeZhuanDianBolt", new ZongHeZhuanBolt(), 1).shuffleGrouping("ZongHeZuHeDianBolt");
        builder.setBolt("ZongHeMagicToDianBolt", new ZongHeMagicToDianBolt(), 1).allGrouping("ZongHeZhuanDianBolt");
        // 气象数据--线
        builder.setBolt("MPGKaZhiXianBolt", new MPGKaZhiBolt(), 1).allGrouping("YaoSuZuHeXianBolt");
        builder.setBolt("MPGShouJiXianBolt", new MPGShouJiBolt(), 1).shuffleGrouping("MPGKaZhiXianBolt");
        builder.setBolt("MPGJiHeToXianDanJiHeBolt", new MPGJiHeDanJiHeBolt(), mpgLine).shuffleGrouping("MPGShouJiXianBolt");
        builder.setBolt("MPGMagicToXianBolt", new MPGMagicToXianBolt(), mpgLineJiHe).shuffleGrouping("MPGJiHeToXianDanJiHeBolt");
        // 气象数据--点
        builder.setBolt("MPGKaZhiDianBolt", new MPGKaZhiBolt(), 1).allGrouping("YaoSuZuHeDianBolt");
        builder.setBolt("MPGShouJiDianBolt", new MPGShouJiBolt(), 1).shuffleGrouping("MPGKaZhiDianBolt");
        builder.setBolt("MPGJiHeToDianDanJiHeBolt", new MPGJiHeDanJiHeBolt(), mpgDian).shuffleGrouping("MPGShouJiDianBolt");
        builder.setBolt("MPGMagicToDianBolt", new MPGMagicToDianBolt(), mpgDianJiHe).shuffleGrouping("MPGJiHeToDianDanJiHeBolt");
        // 写回kafka
        builder.setBolt("ToBeanKafkaBolt", new ToBeanKafkaBolt(), 1).allGrouping("RiskKaZhiXianBolt");
        builder.setBolt("ToRiskLinkKafkaBolt", new ToRiskLinkKafkaBolt(), 1).allGrouping("RiskLineDataBolt");
        builder.setBolt("ToRiskKafkaBolt", new ToRiskKafkaBolt(), 1).allGrouping("RiskMagicToLineBolt").allGrouping("RiskMagicToDianBolt").allGrouping("ZongHeMagicToLineBolt").allGrouping("ZongHeMagicToDianBolt");
        builder.setBolt("ToMPGLineKafkaBolt", new ToMPGLineKafkaBolt(), 1).allGrouping("MPGMagicToXianBolt");
        builder.setBolt("ToMPGPointKafkaBolt", new ToMPGPointKafkaBolt(), 1).allGrouping("MPGMagicToDianBolt");

        // topology系统配置文件的设置
        Config config = new Config();
        config.put(Config.TOPOLOGY_DEBUG, false);
        config.put(KAFKA_PRODUCER_SERVER, properties.getProperty(KAFKA_PRODUCER_SERVER));
        config.put(KAFKA_PRODUCER_BEAN_TOPIC, properties.getProperty(KAFKA_PRODUCER_BEAN_TOPIC));
        config.put(KAFKA_PRODUCER_MPG_Line_TOPIC, properties.getProperty(KAFKA_PRODUCER_MPG_Line_TOPIC));
        config.put(KAFKA_PRODUCER_MPG_Point_TOPIC, properties.getProperty(KAFKA_PRODUCER_MPG_Point_TOPIC));
        config.put(KAFKA_PRODUCER_RISK_TOPIC, properties.getProperty(KAFKA_PRODUCER_RISK_TOPIC));
        config.put(KAFKA_PRODUCER_RISK_LINK_TOPIC, properties.getProperty(KAFKA_PRODUCER_RISK_LINK_TOPIC));
        config.put(TYPE_PRODUCE, properties.getProperty(TYPE_PRODUCE));
        // config.put("pointJson", pointJson.toString());
        // config.put("contentJson", contentJson.toString());
        config.put(REDIS_KAZHI, properties.getProperty(REDIS_KAZHI, "riskKaZhi"));
        config.put(REDIS_POINT_KAZHI, properties.getProperty(REDIS_POINT_KAZHI, "riskIndexOfPoint"));
        config.put(REDIS_MPG_KAZHI, properties.getProperty("redis.key.mpg.kazhi", "climateIndex"));
        config.put(REDIS_RISK_REV, properties.getProperty(REDIS_RISK_REV, "riskIndexRevision"));
        config.put("queryLonLat", properties.getProperty("top.url.queryLonLat"));
        config.put("queryPointSclie", properties.getProperty("top.url.queryPointSclie"));
        config.put("queryLineSclie", properties.getProperty("top.url.queryLineSclie"));
        config.put("queryFilter", properties.getProperty("top.url.queryFilter"));
        config.put("queryPac", properties.getProperty("top.url.queryPac"));
        config.put("queryID", properties.getProperty("top.url.queryID"));
        config.put("queryRiskInfo", properties.getProperty("top.url.queryRiskInfo"));
        // TODO redis连接池配置
        // TODO bolt里的配置获取
        // config.setMaxSpoutPending(Integer.parseInt(properties.getProperty("max.spout.pending")));
        // config.setMessageTimeoutSecs(60);
        config.setNumWorkers(Integer.parseInt(properties.getProperty(TOP_WORKER_NUMBER, DEFAULT_TOP_WORKER_NUMBER)));
        // config.setDebug(false);

        //在本地运行
        if (args.length < 1) {
            new LocalCluster().submitTopology(properties.getProperty(TOPOLOGY_NAME, DEFAULT_TOPOLOGY_NAME), config, builder.createTopology());
        } else {
            try {
                StormSubmitter.submitTopology(properties.getProperty(TOPOLOGY_NAME, DEFAULT_TOPOLOGY_NAME), config, builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                LOGGER.error(e.toString());
                e.printStackTrace();
            }
        }
    }

    //读取配置文件
    public static Properties getProperties(String confPath) {
        InputStream in = TopologyDrive.class.getClassLoader().getResourceAsStream(confPath);
        Properties prop = new Properties();
        try {
            prop.load(in);//获取配置文件
            in.close();
        } catch (IOException e) {
            LOGGER.error("No config.properties defined error");
        }
        return prop;
    }


}